package dli_and_obs.job_utils

import java.util.Properties

import com.huawei.dli.sdk.DLIClient
import com.huawei.dli.sdk.authentication.AuthenticationMode
import com.huawei.dli.sdk.common.DLIInfo
import dli_and_obs.utils
/**
  * @Description TODO 
  * @Author liutaowei
  * @Date 2019-5-22 15:30
  */
object dli_job_log {

  def DLI_joblog(properties: Properties):String={

    val log = log_findAndWrite_util(properties)

    println(log)
    return "success"
  }

  def log_findAndWrite_util(properties:Properties)={
    val client = get_client(properties)
    val log = get_logs(client,properties)

    log
  }

  def get_client(properties:Properties) = {
    val ak = properties.getProperty("ak")
    val sk = properties.getProperty("sk")

    val regionName = properties.getProperty("regionName")
    val projectId = properties.getProperty("projectId")

    val dliInfo = new DLIInfo(regionName, ak, sk, projectId)
    val client = new DLIClient(AuthenticationMode.AKSK, dliInfo)
    client
  }

  def get_logs(client:DLIClient,properties:Properties):String ={
    import scala.collection.JavaConverters._
    val cluster_name = properties.getProperty("cluster_name")

    val query_by = properties.getProperty("query_by")

    var batchJobs = client.listAllBatchJobs(cluster_name,0,30).asScala
      .filter(_.getState.toString == "DEAD")

    println(query_by)
    if(null == query_by){}
    else if(query_by.equals("_name"))
      {batchJobs = batchJobs.filter(_.getName.equals(properties.getProperty("job_name").trim))}
    else if(query_by.equals("_id"))
      {batchJobs = batchJobs.filter(_.getName.equals(properties.getProperty("job_id").trim))}
    else if(query_by.equals("_time")){
        val time = date_to_time(properties.getProperty("job_time")).toString
        batchJobs = batchJobs.filter(_.getCreateTime.toString>=time)
    }


    var log = ""
    if( null!= batchJobs  && batchJobs.size > 0){
      // 根据id找到相关日志

      val job_id = batchJobs(0).getId // 找到id
      try {
        val logs = client.getRestApi.getBatchLog(client.getProjectId, job_id, client.getAuthToken, 0, 1000, "csv", 0)
        val batch = client.getRestApi.getBatchLog(client.getProjectId, job_id, client.getAuthToken, 0, 1000, "csv", 0)
        import scala.collection.JavaConverters._
         log = batch.getLog.asScala.mkString("\n")

//        val name = batchJobs(0).getName
//        properties.setProperty("create_time",batchJobs(0).getCreateTime.toString)
//          println(log)
//        utils.log_utils.log_writeto_obs(properties,log.toString,name)

      }catch {
        case var6: Exception =>
          log ="\nERROR: 所选日志已经清除，请重新选择"
      }
    }

    return log
  }


  def date_to_time(date:String)={
    import java.text.SimpleDateFormat
    val simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd")
    val time = simpleDateFormat.parse(date)
    val ts = time.getTime
    ts
  }
}
